package com.ldw.mq;

import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

public class Producer {
    public static void main(String[] args) throws Exception {
        sendGeneralNews();
        sendSequentialMessage();
        sendTransactionMessage();
        sendSingleMessage();
        sendBatchMessages();
    }

    /**
     * 发送普通消息
     */
    public static void sendGeneralNews() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        producer.setNamesrvAddr("localhost:9876");
        producer.setRetryTimesWhenSendAsyncFailed(2);
        Message msg = new Message("TopicTest", "TagA", "OrderID188",
                "Hello World".getBytes(RemotingHelper.DEFAULT_CHARSET));
        SendResult sendResult = producer.send(msg);
        System.out.printf("%s%n", sendResult);
        producer.shutdown();
    }

    /**
     * 发送顺序消息
     */
    public static void sendSequentialMessage() throws Exception {
        //第一步：初始化生产者，配置生产者参数，启动生产者
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        producer.start();
        //第二步：初始化消息对象
        Message msg = new Message("TopicTest", "消息过滤词", "OrderID188",
                "Hello World".getBytes(RemotingHelper.DEFAULT_CHARSET));
        Integer hashKey = 123;
        //第三步：核心操作MessageQueueSelector，根据hashKey选择当前消息发送到哪个分区中
        SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                Integer id = (Integer) arg;
                int index = id % mqs.size();
                return mqs.get(index);
            }
        }, hashKey);
        producer.shutdown();
    }

    /**
     * 事务消息
     */
    public static void sendTransactionMessage() throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer("groupName");
        //配置生产者的各个参数和Broker回调，检查本地事务处理并启动生产者
//        producer.setCheckThreadPoolMaxSize(2);
//        producer.setCheckThreadPoolMinSize(2);
        producer.setTransactionListener(new TransactionListener() {
            private final AtomicInteger transactionIndex = new AtomicInteger(0);

            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                int value = transactionIndex.getAndIncrement();
                if (value == 0) {
                    throw new RuntimeException("Could find db");
                } else if ((value % 5) == 0) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else if ((value % 4) == 0) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                return LocalTransactionState.UNKNOW;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.printf("server checking TrMsg %s%n", msg);
                int value = transactionIndex.getAndIncrement();
                if ((value % 6) == 0) {
                    throw new RuntimeException("Could find db");
                } else if ((value % 5) == 0) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else if ((value % 4) == 0) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                return LocalTransactionState.UNKNOW;
            }
        });
        producer.start();
        //第三步：设置本地事务处理器，发送消息
        Message msg = new Message("Topic", "tags", "KEY", ("Hello RocketMQ" + System.currentTimeMillis()).getBytes(RemotingHelper.DEFAULT_CHARSET));
        SendResult sendResult = producer.sendMessageInTransaction(msg, null);
        System.out.printf("%s%n", sendResult);
    }

    /**
     * 发送单项消息
     * 单向消息的生产者只管发送过程，不管发送结果。单项消息主要用于日志传输等消息允许丢失的场景，
     */
    public static void sendSingleMessage() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        producer.setNamesrvAddr("127.0.0.1:9876;127.0.0.2:9872");
        producer.setInstanceName("instance name");
        producer.start();
        Message msg = new Message("TopicTest", "TagA", "OrderID188",
                "Hello World".getBytes(RemotingHelper.DEFAULT_CHARSET));
        producer.sendOneway(msg);
    }

    /**
     * 批量消息发送
     */
    public static void sendBatchMessages() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("BatchProducerGroupName");
        producer.start();
        String topic = "BatchTest";
        List<Message> messages = new ArrayList<>();
        messages.add(new Message(topic, "Tag", "OrderID001", "Hello World0".getBytes()));
        messages.add(new Message(topic, "Tag", "OrderID002", "Hello World2".getBytes()));
        messages.add(new Message(topic, "Tag", "OrderID003", "Hello World3".getBytes()));
        producer.send(messages);
        producer.shutdown();
    }
}
